package main

import (
	"fmt"
	"strings"
	"time"

	"github.com/Shopify/sarama"
	"github.com/kpango/glg"
)

type KafkaHandler struct {
	offset   int64
	producer sarama.AsyncProducer
	closed   chan interface{}
}

func NewKafkaHandler() (KafkaHandler, error) {
	handler := KafkaHandler{}
	handler.closed = make(chan interface{})
	producerConfig := sarama.NewConfig()
	producerConfig.Producer.Return.Successes = true
	producerConfig.Producer.Timeout = 60 * time.Second
	producer, err := sarama.NewAsyncProducer(strings.Split(*KUrl, ","), producerConfig)

	if err != nil {
		return handler, fmt.Errorf("storeService.NewKafkaHandler: Failed to create Kafka producer: %v", err)
	}
	handler.producer = producer
	return handler, nil
}

func (kh *KafkaHandler) Close() error {
	go func() {
		kh.closed <- true
	}()
	if err := kh.producer.Close(); err != nil {
		return err
	}
	return nil
}

func (kh *KafkaHandler) Produce(appInfo, data string) (int64, error) {
	msg := &sarama.ProducerMessage{
		Timestamp: time.Now(),
		Topic:     *KTopic,
		Key:       sarama.StringEncoder(appInfo),
		Value:     sarama.StringEncoder(data),
	}
	go func() {
		kh.producer.Input() <- msg
	}()
	select {
	case err := <-kh.producer.Errors():
		if err != nil {
			logmsg := "storeService.Produce: Failed to produce message to Kafka. Topic:%v, Key:%v, Value:%v"
			glg.Errorf(logmsg, msg.Topic, msg.Key, msg.Value)
			return kh.offset, err
		}
	case msg := <-kh.producer.Successes():
		key, err := msg.Key.Encode()
		if err != nil {
			return kh.offset, err
		}
		val, err := msg.Value.Encode()
		if err != nil {
			return kh.offset, err
		}
		kh.offset = msg.Offset
		logmsg := "storeService.Produce: Produced message successfully to Kafka. Topic:%v, Key:%v, Value:%v"
		glg.Infof(logmsg, msg.Topic, string(key), string(val))
	}
	return kh.offset, nil
}
